#include "pch.h"
#include "ThreadpoolWork.h"
#include "AsyncLibImpl.h"

namespace AsyncLib
{

CThreadpoolWork::CThreadpoolWork(UINT nMinThreads, UINT nMaxThreads) : m_nMinThreads(nMinThreads), m_nMaxThreads(nMaxThreads), m_nCurThreads(0), m_pTpWork(NULL)
{
}

CThreadpoolWork::~CThreadpoolWork()
{
	assert(m_pTpWork == NULL);
	assert(m_deqTasks.empty());
}

void CThreadpoolWork::Init()
{
	m_pTpWork = CreateThreadpoolWork([](PTP_CALLBACK_INSTANCE, PVOID pContext, PTP_WORK)
	{
		((CThreadpoolWork *)pContext)->OnCallback();
	}, this, g_asyncLib.GetCallbackEnv());

	assert(m_pTpWork);
}

void CThreadpoolWork::Stop()
{
	std::unique_lock lock(m_mutex);
}

void CThreadpoolWork::Exit()
{
	// It has been waited and closed outside
	m_pTpWork = NULL;

	// Destruct pending tasks sequentially
	while (m_deqTasks.size())
	{
		m_deqTasks.pop_front();
	}
}

void CThreadpoolWork::PostTask(std::function<void()>&& fnTask)
{
	std::unique_lock lock(m_mutex);
	m_deqTasks.emplace_back(std::move(fnTask));

	// Less than the number of core threads, or the number of tasks is more than half of the number of threads, start a new thread
	if (m_nCurThreads < m_nMinThreads || m_nCurThreads < m_nMaxThreads && m_nCurThreads < m_deqTasks.size() * 2)
	{
		m_nCurThreads++;
		SubmitThreadpoolWork(m_pTpWork);
	}
}

void CThreadpoolWork::OnCallback()
{
	g_asyncLib.SetInThreadPool();
	std::unique_lock lock(m_mutex);

	while (g_asyncLib.IsRunning() && m_deqTasks.size())
	{
		auto fnTask = std::move(m_deqTasks.front());
		m_deqTasks.pop_front();
		lock.unlock();

		fnTask();
		fnTask = nullptr;

		lock.lock();
	}

	m_nCurThreads--;
}

} // namespace AsyncLib
